package tk.xboot.kafka.kfk;

import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.util.Optional;

@Slf4j
@Component
public class Listener {

    //@KafkaListener(topics= "WordsWithCountsTopic", groupId = "myGroup")
    public void processMessage(String content) {
        // TODO
        log.info(">>>> received message:{} from topic testTopic", content);
        System.out.println(">>>> received message:{} from topic testTopic");
    }


    @KafkaListener(topics = "${ws.kfk.topic}", groupId = "myGroupx")
    public void listen(ConsumerRecord<?, ?> record) {
        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        log.info(">>>>>>>>>>>>>>>>>>>>>");

        if (kafkaMessage.isPresent()) {
            Object message = kafkaMessage.get();
            try {
                log.info(message.toString());
            } catch (Exception e) {
                log.error("异常, 消息:{}, 异常:{}", message.toString(), e);
            }
        }
    }
}

